package com.shujia.dwi

import com.shujia.common.SparkTool
import org.apache.spark.sql.expressions.{UserDefinedFunction, Window}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import com.shujia.common.utils.Geography
import org.slf4j.Logger

object DwiResRegnMergelocationMskDay extends SparkTool {
  // 构建位置数据融合表

  /**
   * 将代码打包并提交到集群运行：
   * 1、需要使用ctyun模块进行打包
   * 2、由于dwi中使用了common中的代码，所以需要将common的jar包一起上传
   * 3、提交命令：
   * spark-submit --conf spark.sql.shuffle.partitions=10 --master yarn --deploy-mode client --jars common-1.0.jar --class com.shujia.dwi.DwiResRegnMergelocationMskDay  dwi-1.0.jar 20231027
   */

    override def run(spark:SparkSession): Unit = {
        // 基于用户位置数据，通过对开始时间进行排序，再计算相邻两条位置记录之间的距离差、时间差
        // 并推导计算出位置移动的速度，然后进行超速乒乓清洗
        // 距离差：通过两组经纬度计算
        // 时间差：通过后一条记录的开始时间-上一条数据的结束时间

        import spark.implicits._
        import org.apache.spark.sql.functions._

        // 将普通的方法转换成UDF方法
        val CalculateLengthWithLgLat: UserDefinedFunction = udf((longi: Double, lati: Double, last_longi: Double, last_lati: Double) => {
          Geography.calculateLength(longi, lati, last_longi, last_lati)
        })

        // 从Hive的ods层中加载ods_oidd表的数据
        val oiddDF: DataFrame = spark.table("ods.ods_oidd").where($"day_id" === s"$dayId")

        val dwiResRegnMergelocationMskDayDF:DataFrame = oiddDF
          // 将空的数据过滤掉
          .where($"mdn" =!= "")
          // 将时间拆分成开始时间、结束时间
          .withColumn("start_t", split($"start_time", ",")(1))
          .withColumn("end_t", split($"start_time", ",")(0))
          /**
           * 在时间线上进行聚类：为了解决同一个网格内同一个用户的多条连续的位置记录有时间交叠问题
           */
          // 按照用户进行partitionBy，按开始时间进行orderBy排序，取上一条数据的网格ID
          .withColumn("last_grid_id", lag($"grid_id", 1) over Window.partitionBy($"mdn").orderBy($"start_t"))
          // 对比当前记录的网格id同上一条数据的网格id是否一致，如果一样则置为0，不一样则置为1
          .withColumn("flag", when($"grid_id" === $"last_grid_id", 0).otherwise(1))
          // 对flag这一列进行累加求和得到grp列
          .withColumn("grp", sum($"flag") over Window.partitionBy($"mdn").orderBy($"start_t"))
          // 按照grp分组，取最小的时间作为开始时间，取最大的时间作为结束时间，对数据进行合并
          .groupBy($"grp", $"mdn", $"grid_id", $"county_id", $"longi", $"lati", $"bsid")
          .agg(min($"start_t") as "start_date", max($"end_t") as "end_date")

          // 计算速度进行超速乒乓清洗
          // 取相邻两条记录的前一条数据的开始时间：避免不同网格间相邻位置记录存在时间交叠问题
          .withColumn("last_start_date", lag($"start_date", 1) over Window.partitionBy($"mdn").orderBy($"start_date"))
          // 计算时间差，单位：秒
          .withColumn("diff_time", unix_timestamp($"start_date", "yyyyMMddHHmmss") - unix_timestamp($"last_start_date", "yyyyMMddHHmmss"))
          // 取相邻两条记录的前一条数据的经纬度
          .withColumn("last_longi", lag($"longi", 1) over Window.partitionBy($"mdn").orderBy($"start_date"))
          .withColumn("last_lati", lag($"lati", 1) over Window.partitionBy($"mdn").orderBy($"start_date"))
          // 通过经纬度计算距离
          .withColumn("distance", CalculateLengthWithLgLat($"longi", $"lati", $"last_longi", $"last_lati"))
          // 计算速度
          .withColumn("speed", round($"distance" / $"diff_time", 4))
          // 基于速度进行超速乒乓清洗
          /**
           * 一般出行方式的速度，单位：m/s
           * 1、步行：1~3m/s
           * 2、自行车：5m/s
           * 3、摩托车：15~20m/s
           * 4、汽车：15~30m/s
           * 5、火车：20~35m/s
           * 6、动车：30~50m/s
           * 7、高铁：75m/s
           * 8、飞机：200m/s
           */
          .where($"speed" < 300)
          // 整理数据
          .select(
            upper(md5($"mdn")) as "mdn" // 对手机号进行脱敏处理
            , $"start_date"
            , $"end_date"
            , $"county_id"
            , $"longi"
            , $"lati"
            , $"bsid"
            , $"grid_id"
          )

            // 将结果数据写入对应的结果目录
            dwiResRegnMergelocationMskDayDF
              .write
              .format("csv")
              .option("sep",",")
              .mode(SaveMode.Overwrite)
              .save(s"/daas/motl/dwi/dwi_res_regn_mergelocation_msk_d/day_id=$dayId")

        // 给表增加分区
        spark.sql(
          s"""
            |alter table dwi.dwi_res_regn_mergelocation_msk_d add if not exists partition(day_id=$dayId)
            |""".stripMargin)


      }


}
